﻿/*
-----------------------------------------------------------------------------
This source file is part of Cell Cloud.

Copyright (c) 2009-2012 Cell Cloud Team - cellcloudproject@gmail.com

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
-----------------------------------------------------------------------------
*/

#include "CellStableHeaders.h"
#include "CellBlockingConnector.h"
#include "CellBlockingConnectorThread.h"
#include "CellSession.h"
#include "CellMessage.h"
#include "CellInet4Address.h"
#include "CellInet6Address.h"
#include "CellLoggerManager.h"
#include "CellStringConverter.h"

namespace cell
{

#define BC_DEFAULT_BUFSIZE 8192
#define BC_CONNECT_TIMEOUT 15000

//-----------------------------------------------------------------------
BlockingConnector::BlockingConnector(void)
	: mSession(NULL)
	, mRecvThread(NULL)
	, mBufSize(BC_DEFAULT_BUFSIZE)
	, mConnectTimeout(BC_CONNECT_TIMEOUT)
	, mSock(INVALID_SOCKET)
{
}
//-----------------------------------------------------------------------
BlockingConnector::~BlockingConnector(void)
{
	disconnect();
}
//-----------------------------------------------------------------------
Session* BlockingConnector::connect(const String& addr, uint port)
{
	Inet4Address address(addr, port);
	return connect(address);
}
//-----------------------------------------------------------------------
Session* BlockingConnector::connect(const InetAddress& address)
{
	{
		CELL_LOCK_MUTEX(mMutex);

		if (NULL != mSession)
		{
			fireErrorOccurred(MessageHandler::EC_STATE_ERROR);
			return NULL;
		}

		if (!address.isValid())
		{
			fireErrorOccurred(MessageHandler::EC_ADDRESS_INVALID);
			return NULL;
		}

		// 创建套接字
		if (INVALID_SOCKET == (mSock = ::socket(AF_INET, SOCK_STREAM, 0)))
		{
			fireErrorOccurred(MessageHandler::EC_SOCK_FAILED);
			return NULL;
		}

		// 设置超时
		int timeOut = 30000;
		if(::setsockopt(mSock, SOL_SOCKET, SO_SNDTIMEO, (char *)&timeOut, sizeof(timeOut)) < 0)
		{
			closeSocket(mSock);
			mSock = INVALID_SOCKET;
			fireErrorOccurred(MessageHandler::EC_SOCK_FAILED);
			return NULL;
		}
		timeOut = 30000;
		if (::setsockopt(mSock, SOL_SOCKET, SO_RCVTIMEO, (char *)&timeOut, sizeof(timeOut)) < 0)
		{
			closeSocket(mSock);
			mSock = INVALID_SOCKET;
			fireErrorOccurred(MessageHandler::EC_SOCK_FAILED);
			return NULL;
		}

		// 设置缓冲区
		/*int bufSize = mBufSize + 1024;
		if (::setsockopt(mSock, SOL_SOCKET, SO_RCVBUF, (char *)&bufSize, sizeof(bufSize)) == SOCKET_ERROR)
		{
			LoggerManager::getSingleton().log("[CC] Sets socket revbuf failed.");
		}
		bufSize = mBufSize + 1024;
		if (::setsockopt(mSock, SOL_SOCKET, SO_SNDBUF, (char *)&bufSize, sizeof(bufSize)) == SOCKET_ERROR)
		{
			LoggerManager::getSingleton().log("[CC] Sets socket sndbuf failed.");
		}*/

		// Keep alive
		int keepAlive = 1;
		if (::setsockopt(mSock, SOL_SOCKET, SO_KEEPALIVE, (char *)&keepAlive, sizeof(keepAlive)) < 0)
		{
			closeSocket(mSock);
			mSock = INVALID_SOCKET;
			fireErrorOccurred(MessageHandler::EC_SOCK_FAILED);
			return NULL;
		}

		// 设置非阻塞方式连接   
		unsigned long argp = 1;
		if (::ioctlsocket(mSock, FIONBIO, (unsigned long*)&argp) < 0)
		{
			closeSocket(mSock);
			mSock = INVALID_SOCKET;
			fireErrorOccurred(MessageHandler::EC_SOCK_FAILED);
			return NULL;
		}

		unsigned long destAddr = inet_addr(address.getHostAddress().c_str());
		SOCKADDR_IN destSockAddr;
		memcpy(&destSockAddr.sin_addr, &destAddr, sizeof(destAddr));
		destSockAddr.sin_family = AF_INET;
		destSockAddr.sin_port = htons(address.getPort());
		destSockAddr.sin_addr.s_addr = inet_addr(address.getHostAddress().c_str());

		// 连接
		::connect(mSock, (LPSOCKADDR)&destSockAddr, sizeof(destSockAddr));
	}

	struct fd_set fds;
	struct timeval tv;
	tv.tv_sec = (int)(mConnectTimeout / 1000);
	tv.tv_usec = 0;
	FD_ZERO(&fds);
	FD_SET(mSock, &fds);
	int result = ::select(0, NULL, &fds, NULL, &tv);
	if (result <= 0)
	{
		closeSocket(mSock);
		mSock = INVALID_SOCKET;
		fireErrorOccurred(MessageHandler::EC_CONNECT_TIMEOUT);
		return NULL;
	}

	// 设置回阻塞模式
	unsigned long argp_b = 0;
	if (::ioctlsocket(mSock, FIONBIO, (unsigned long*)&argp_b) < 0)
	{
		closeSocket(mSock);
		mSock = INVALID_SOCKET;
		fireErrorOccurred(MessageHandler::EC_SOCK_FAILED);
		return NULL;
	}

	// 创建 Session
	fireSessionCreated(address.getHostAddress().c_str(), address.getPort());

	// 启动线程
	mRecvThread = CELL_NEW BlockingConnectorThread(this);
	mRecvThread->start();

	return mSession;
}
//-----------------------------------------------------------------------
void BlockingConnector::disconnect(void)
{
	if (mSock != INVALID_SOCKET)
	{
		if (mRecvThread != NULL)
		{
			// 停止线程
			mRecvThread->stop();
		}

		// 关闭 Socket 句柄
		closeSocket(mSock);
		mSock = INVALID_SOCKET;
	}

	// 等待线程结束
	if (mRecvThread != NULL)
	{
		mRecvThread->stop();

		while (mRecvThread->isRunning())
		{
			Thread::sleep(10);
		}

		CELL_DELETE mRecvThread;
		mRecvThread = NULL;
	}

	{
		CELL_LOCK_MUTEX(mMutex);

		if (mSession != NULL)
		{
			CELL_DELETE mSession;
			mSession = NULL;
		}
	}

	while (false == mWriteQueue.empty())
	{
		char* buf = NULL;
		if (mWriteQueue.poll(buf))
		{
			CELL_FREE(buf, MEMCATEGORY_GENERAL);
		}
	}
}
//-----------------------------------------------------------------------
inline void BlockingConnector::setConnectTimeout(int milliseconds)
{
	mConnectTimeout = milliseconds;
}
//-----------------------------------------------------------------------
FORCEINLINE Session* BlockingConnector::getSession(void) const
{
	return mSession;
}
//-----------------------------------------------------------------------
inline bool BlockingConnector::isConnected(void)
{
	CELL_LOCK_MUTEX(mMutex);
	return (NULL != mSession);
}
//-----------------------------------------------------------------------
void BlockingConnector::write(Message* message)
{
	uint32 dataSize = message->length();
	if (dataSize == 0)
	{
		// 通知错误
		fireErrorOccurred(MessageHandler::EC_WRITE_FAILED);
		return;
	}

	// 获取数据
	char* data = CELL_ALLOC_T(char, dataSize + 1, MEMCATEGORY_GENERAL);
	memset(data, 0x0, dataSize + 1);
	message->get(data);

	// 数据缓存
	char* buf = NULL;
	int size = 0;

	if (existDataMark())
	{
		size = dataSize + getHeadLength() + getTailLength();
		buf = CELL_ALLOC_T(char, size + 1, MEMCATEGORY_GENERAL);
		memset(buf, 0x0, size + 1);
		char* cursor = buf;
		memcpy(cursor, getHeadMark(), getHeadLength());
		cursor += getHeadLength();
		memcpy(cursor, data, dataSize);
		cursor += dataSize;
		memcpy(cursor, getTailMark(), getTailLength());
	}
	else
	{
		size = dataSize;
		buf = CELL_ALLOC_T(char, size + 1, MEMCATEGORY_GENERAL);
		memset(buf, 0x0, size + 1);
		memcpy(buf, data, size);
	}

	CELL_FREE(data, MEMCATEGORY_GENERAL);

	bool error = false;

	{
		CELL_LOCK_MUTEX(mMutex);

		mWriteQueue.offer(buf);

		while (false == mWriteQueue.empty())
		{
			char* databuf;
			mWriteQueue.peek(databuf);

			int ssize = 0;
			int ss = 0;
			while (ssize < size)
			{
				if (SOCKET_ERROR == (ss = ::send(mSock, databuf + ssize, size - ssize, 0)))
				{
					error = true;
					break;
				}

				ssize += ss;
			}

			if (false == error)
			{
				char* dbuf;
				mWriteQueue.poll(dbuf);
				CELL_FREE(dbuf, MEMCATEGORY_GENERAL);
			}
			else
			{
				break;
			}
		}
	}

	if (false == error)
	{
		fireMessageSent(message);
	}
	else
	{
		fireErrorOccurred(MessageHandler::EC_WRITE_FAILED);
	}
}
//-----------------------------------------------------------------------
void BlockingConnector::write(Session* session, Message* message)
{
	{
		CELL_LOCK_MUTEX(mMutex);

		if (NULL == mSession || (session->getId() != mSession->getId()))
		{
			// 通知错误
			fireErrorOccurred(MessageHandler::EC_WRITE_FAILED);
			return;
		}
	}

	write(message);
}
//-----------------------------------------------------------------------
void BlockingConnector::read(Message* message, Session* session)
{
	// TODO
}
//-----------------------------------------------------------------------
void BlockingConnector::fireSessionCreated(const char* host, const uint port)
{
	// 获取本机地址
	/*
	PHOSTENT hostinfo;
	char name[255] = {0x0};
	String ip;
	if (gethostname(name, sizeof(name)) == 0)
	{
		if((hostinfo = gethostbyname(name)) != NULL)
		{
			ip = inet_ntoa(*(struct in_addr *)*hostinfo->h_addr_list);
		}
	}
	Inet4Address address(ip);
	*/

	Inet4Address address(host, port);

	{
		CELL_LOCK_MUTEX(mMutex);

		// 创建 Session
		mSession = CELL_NEW Session(this, address);
	}

	if (NULL != mHandler)
	{
		mHandler->sessionCreated(mSession);
	}
}
//-----------------------------------------------------------------------
void BlockingConnector::fireSessionDestroy(void)
{
	if (NULL != mHandler && NULL != mSession)
	{
		mHandler->sessionDestroyed(mSession);
	}

	CELL_LOCK_MUTEX(mMutex);

	if (mSession != NULL)
	{
		CELL_DELETE mSession;
		mSession = NULL;
	}
}
//-----------------------------------------------------------------------
void BlockingConnector::fireSessionOpened(void)
{
	if (NULL != mHandler)
	{
		mHandler->sessionOpened(mSession);
	}
}
//-----------------------------------------------------------------------
void BlockingConnector::fireSessionClosed(void)
{
	if (NULL != mHandler)
	{
		mHandler->sessionClosed(mSession);
	}

	if (INVALID_SOCKET != mSock)
	{
		closeSocket(mSock);
		mSock = INVALID_SOCKET;
	}

	fireSessionDestroy();
}
//-----------------------------------------------------------------------
void BlockingConnector::fireMessageReceived(Message* msg)
{
	if (NULL != mHandler)
	{
		mHandler->messageReceived(mSession, msg);
	}
}
//-----------------------------------------------------------------------
void BlockingConnector::fireMessageSent(Message* msg)
{
	if (NULL != mHandler)
	{
		mHandler->messageSent(mSession, msg);
	}
}
//-----------------------------------------------------------------------
void BlockingConnector::fireErrorOccurred(uint16 errorCode)
{
	if (NULL != mHandler)
	{
		mHandler->errorOccurred(errorCode, mSession);
	}
}
//-----------------------------------------------------------------------
FORCEINLINE socket_t BlockingConnector::getSock(void)
{
	return mSock;
}

} // end namespace cell
